package cn.doitedu.olap_agg;

import cn.doitedu.udfs.TimeStampFloorUDF;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/8
 * @Desc: 学大数据，上多易教育
 *   流量主题 olap分析所需的 轻度聚合表 计算
 **/
public class Job01_TrafficOlapAggregate {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().set("table.exec.source.idle-timeout","1 ms");


        // 建表，映射源表（ dwd层的维度退化表- 用户行为日志明细 ）
        tenv.executeSql(
                "create table dwd_events_kafka(     " +
                        "     release_channel    string             " +
                        "    ,device_type        string             " +
                        "    ,session_id         string             " +
                        "    ,event_id           string             " +
                        "    ,action_time        bigint             " +
                        "    ,properties         map<string,string> " +
                        "    ,url as   properties['url']            " +
                        "    ,user_id bigint             " +
                        "    ,create_time timestamp(3)  " +
                        "    ,province   string        " +
                        "    ,city       string        " +
                        "    ,region     string        " +
                        "    ,page_type       string   " +
                        "    ,page_service    string   " +
                        "    ,rt as to_timestamp_ltz(action_time,3)   " +
                        "    ,watermark for rt as rt - interval '0' second  " +

                        ") WITH (                      " +
                        "    'connector' = 'kafka',    " +
                        "    'topic' = 'dwd-user-action-log',  " +
                        "    'properties.bootstrap.servers' = 'doitedu:9092',  " +
                        "    'properties.group.id' = 'doit44_g1',              " +
                        "    'scan.startup.mode' = 'latest-offset',            " +
                        "    'value.format' = 'json',                          " +
                        "    'value.fields-include' = 'EXCEPT_KEY'             " +
                        ")                                                     ");



        // 建表，映射目标表（doris中dws层的 聚合表  dws.olap_traffic_agg_01 ）
        tenv.executeSql(
                        " create table doris_sink(                                                                            "+
                        "     dt                 date                                                                         "+
                        "     ,m60_time          string                                                                       "+
                        "     ,m30_time          string                                                                       "+
                        "     ,m10_time          string                                                                       "+
                        "     ,m05_time          string                                                                       "+
                        "     ,m01_time          string                                                                       "+
                        "     ,province          string                                                                       "+
                        "     ,city              string                                                                       "+
                        "     ,region            string                                                                       "+
                        "     ,device_type       string                                                                       "+
                        "     ,release_channel   string                                                                       "+
                        "     ,url               string                                                                       "+
                        "     ,page_type         string                                                                       "+
                        "     ,page_service      string                                                                       "+
                        "     ,is_new            int                                                                          "+

                        "     ,pv                bigint                                                                       "+

                        // -- 名义上的bitmap，实质是： user_id
                        "     ,uv_bitmap         bigint                                                                       "+

                        // -- 名义上的hll，实质是：session_id
                        "     ,ses_hll           string                                                                       "+

                        "  ) WITH (                                                                                           "+
                        "     'connector' = 'doris',                                                                          "+
                        "     'fenodes' = 'doitedu:8030',                                                                     "+
                        "     'table.identifier' = 'dws.olap_traffic_agg_01',                                                 "+
                        "     'username' = 'root',                                                                            "+
                        "     'password' = 'root',                                                                            "+
                        "     'sink.label-prefix' = 'doris_label-47',                                                         "+
                        "     'sink.properties.columns' = 'dt,m60_time,m30_time,m10_time,m05_time,m01_time,province,          "+
                        "     city,region,device_type,release_channel,url,page_type,page_service,is_new,pv,uv_bitmap,         "+
                        "     ses_hll,uv_bitmap=to_bitmap(uv_bitmap),ses_hll=hll_hash(ses_hll) '                              "+
                        "  )                                                                                                  "
        );



        tenv.createTemporaryFunction("ts_floor", TimeStampFloorUDF.class);
        tenv.executeSql(
                        "INSERT INTO  doris_sink   " +
                        "SELECT                                                                                    "+
                        "     to_date(date_format(window_start,'yyyy-MM-dd')) as dt                                 "+
                        "     ,ts_floor(window_start,60) as m60_time                                                "+
                        "     ,ts_floor(window_start,30) as m30_time                                                "+
                        "     ,ts_floor(window_start,10) as m10_time                                                "+
                        "     ,ts_floor(window_start,5)  as m05_time                                                "+
                        "     ,cast(window_start as string) as m01_time                                             "+
                        "     ,province                                                                             "+
                        "     ,city                                                                                 "+
                        "     ,region                                                                               "+
                        "     ,device_type                                                                          "+
                        "     ,release_channel                                                                      "+
                        "     ,url                                                                                  "+
                        "     ,page_type                                                                            "+
                        "     ,page_service                                                                         "+
                        "     ,if(to_date(date_format(create_time,'yyyy-MM-dd'))=current_date,1,0) as is_new        "+
                        "                                                                                           "+
                        "     ,count(event_id) filter(where event_id = 'page_load') as pv                           "+
                        "     ,user_id  as uv_bitmap                                                                "+
                        "     ,session_id as ses_hll                                                                "+
                        "                                                                                           "+
                        " from table(                                                                               "+
                        "     tumble(table dwd_events_kafka,descriptor(rt),interval '1' minute)                     "+
                        " )                                                                                         "+
                        " group by                                                                                  "+
                        "     window_start                                                                          "+
                        "     ,window_end                                                                            "+
                        "     ,province                                                                             "+
                        "     ,city                                                                                 "+
                        "     ,region                                                                               "+
                        "     ,device_type                                                                          "+
                        "     ,release_channel                                                                      "+
                        "     ,url                                                                                  "+
                        "     ,page_type                                                                            "+
                        "     ,page_service                                                                         "+
                        "     ,if(to_date(date_format(create_time,'yyyy-MM-dd'))=current_date,1,0)                  "+
                        "     ,user_id                                                                              "+
                        "     ,session_id                                                                           "
        );

    }
}
